package org.timewindow

import org.FlinkStreamApp
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.bean.MinMaxTemp
import org.diysource.SensorSource

/**
 * description ：增量聚合函数和全窗口聚合函数的结合使用
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/23 10:47
 * modified By ：
 * version:    : 1.0
 */
//  计算窗口中最小温度和最大温度，附加上窗口结束时间
object AggregateWithProcessWindowExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 获取源数据
    val source = environment.addSource(new SensorSource)

    // 计算
    val minTempStream = source
      .map(l => (l.id, l.temperature))
      .keyBy(_._1)
      .timeWindow(Time.seconds(15))
      .aggregate(new MyAgg, new AssignWindowEndProcessFunction)

    // 输出结果
    minTempStream.print
  }

  // IN, ACC, OUT
  class MyAgg extends AggregateFunction[(String, Double), (String, Double, Double), (String, Double, Double)] {
    override def createAccumulator() = ("", Double.MaxValue, Double.MinValue)

    override def add(value: (String, Double), accumulator: (String, Double, Double)) =
      (value._1, accumulator._2.min(value._2), accumulator._2.max(value._2))

    override def getResult(accumulator: (String, Double, Double)) = (accumulator._1, accumulator._2, accumulator._3)

    override def merge(a: (String, Double, Double), b: (String, Double, Double)) = (a._1, a._3.min(b._3), a._2.max(b._2))
  }

  // IN, OUT, KEY, W <: Window]
  // 输入就是 MyAgg 的输出
  class AssignWindowEndProcessFunction extends ProcessWindowFunction[(String, Double, Double), MinMaxTemp, String, TimeWindow] {
    override def process(key: String, context: Context, elements: Iterable[(String, Double, Double)], out: Collector[MinMaxTemp]) = {
      // 由于这里之前使用了 MyAgg 函数，所以返回结果就有一条，每个id, 都返回了 min max 的值，在这里只需要将第一条取出来，把窗口时间加上去就可以了
      val e = elements.iterator.next()
      out.collect(MinMaxTemp(key, e._2, e._3, context.window.getEnd))
    }
  }

}
